//package com.example.user.config;
//
//import com.alibaba.otter.canal.client.CanalConnector;
//import com.alibaba.otter.canal.client.CanalConnectors;
//import com.alibaba.otter.canal.protocol.CanalEntry;
//import com.alibaba.otter.canal.protocol.Message;
//import com.example.user.entity.Users;
//import com.google.protobuf.ByteString;
//import com.google.protobuf.InvalidProtocolBufferException;
//import lombok.extern.slf4j.Slf4j;
//import org.elasticsearch.client.ElasticsearchClient;
//import org.springframework.stereotype.Component;
//import org.springframework.util.CollectionUtils;
//
//import javax.annotation.Resource;
//import java.io.IOException;
//import java.net.InetSocketAddress;
//import java.time.LocalDate;
//import java.time.LocalDateTime;
//import java.time.format.DateTimeFormatter;
//import java.util.ArrayList;
//import java.util.Arrays;
//import java.util.List;
//import java.util.concurrent.TimeUnit;
//
//@Slf4j
//@Component
//public class CanalClient {
//
//    @Resource
//    private ElasticsearchClient client;
//
//
//    /**
//     * 实时数据同步程序
//     *
//     * @throws InterruptedException
//     * @throws InvalidProtocolBufferException
//     */
//    public void run() throws InterruptedException, IOException {
//        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(
//                "localhost", 11111), "", "", "");
//
//        while (true) {
//            //连接
//            connector.connect();
//            //订阅数据库
//            connector.subscribe("seata-users.users");
//            //获取数据
//            Message message = connector.get(100);
//
//            List<CanalEntry.Entry> entryList = message.getEntries();
//            if (CollectionUtils.isEmpty(entryList)) {
//                //没有数据，休息一会
//                TimeUnit.SECONDS.sleep(2);
//            } else {
//                for (CanalEntry.Entry entry : entryList) {
//                    //获取类型
//                    CanalEntry.EntryType entryType = entry.getEntryType();
//
//                    //判断类型是否为ROWDATA
//                    if (CanalEntry.EntryType.ROWDATA.equals(entryType)) {
//                        //获取序列化后的数据
//                        ByteString storeValue = entry.getStoreValue();
//                        //反序列化数据
//                        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
//                        //获取当前事件操作类型
//                        CanalEntry.EventType eventType = rowChange.getEventType();
//                        //获取数据集
//                        List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();
//
//                        if (eventType == CanalEntry.EventType.INSERT) {
//                            log.info("------新增操作------");
//
//                            List<Users> UsersList = new ArrayList<>();
//                            for (CanalEntry.RowData rowData : rowDataList) {
//                                UsersList.add(createUser(rowData.getAfterColumnsList()));
//                            }
//                            //es批量新增文档
//                            index(UsersList);
//                            //打印新增集合
//                            log.info(Arrays.toString(UsersList.toArray()));
//                        } else if (eventType == CanalEntry.EventType.UPDATE) {
//                            log.info("------更新操作------");
//
//                            List<Users> beforeUsersList = new ArrayList<>();
//                            List<Users> afterUsersList = new ArrayList<>();
//                            for (CanalEntry.RowData rowData : rowDataList) {
//                                //更新前
//                                beforeUsersList.add(createUser(rowData.getBeforeColumnsList()));
//                                //更新后
//                                afterUsersList.add(createUser(rowData.getAfterColumnsList()));
//                            }
//                            //es批量更新文档
//                            index(afterUsersList);
//                            //打印更新前集合
//                            log.info("更新前：{}", Arrays.toString(beforeUsersList.toArray()));
//                            //打印更新后集合
//                            log.info("更新后：{}", Arrays.toString(afterUsersList.toArray()));
//                        } else if (eventType == CanalEntry.EventType.DELETE) {
//                            //删除操作
//                            log.info("------删除操作------");
//
//                            List<String> idList = new ArrayList<>();
//                            for (CanalEntry.RowData rowData : rowDataList) {
//                                for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
//                                    if("id".equals(column.getName())) {
//                                        idList.add(column.getValue());
//                                        break;
//                                    }
//                                }
//                            }
//                            //es批量删除文档
//                            delete(idList);
//                            //打印删除id集合
//                            log.info(Arrays.toString(idList.toArray()));
//                        }
//                    }
//                }
//            }
//        }
//    }
//
//    /**
//     * 根据canal获取的数据创建Users对象
//     *
//     * @param columnList
//     * @return
//     */
//    private Users createUser(List<CanalEntry.Column> columnList) {
//        Users user = new Users();
//        DateTimeFormatter dateFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
//        DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
//
//        for (CanalEntry.Column column : columnList) {
//            switch (column.getName()) {
//                case "id" -> user.setId(Integer.valueOf(column.getValue()));
//                case "username" -> user.setUsername(column.getValue());
//                case "password" -> user.setPassword(column.getValue());
//                case "email" -> user.setEmail(column.getValue());
//                case "phone" -> user.setPhone(column.getValue());
//                case "avatar_url" -> user.setAvatarUrl(column.getValue());
//                case "gender" -> user.setGender(column.getValue());
//                case "birthday" -> user.setBirthday(column.getValue().isEmpty() ? null : LocalDate.parse(column.getValue(), dateFormatter));
//                case "created_at" -> user.setCreatedAt(LocalDateTime.parse(column.getValue(), dateTimeFormatter));
//                case "updated_at" -> user.setUpdatedAt(LocalDateTime.parse(column.getValue(), dateTimeFormatter));
//                case "last_login" -> user.setLastLogin(column.getValue().isEmpty() ? null : LocalDateTime.parse(column.getValue(), dateTimeFormatter));
//                case "status" -> user.setStatus("1".equals(column.getValue()));
//                case "score" -> user.setScore(Integer.valueOf(column.getValue()));
//                case "login_count" -> user.setLoginCount(Integer.valueOf(column.getValue()));
//                case "registration_ip" -> user.setRegistrationIp(column.getValue());
//                case "last_login_ip" -> user.setLastLoginIp(column.getValue());
//                case "activation_code" -> user.setActivationCode(column.getValue());
//                case "activation_expiration" -> user.setActivationExpiration(column.getValue().isEmpty() ? null : LocalDateTime.parse(column.getValue(), dateTimeFormatter));
//                case "is_subscribed" -> user.setIsSubscribed("1".equals(column.getValue()));
//                case "remember_token" -> user.setRememberToken(column.getValue());
//                case "deleted_at" -> user.setDeletedAt(column.getValue().isEmpty() ? null : LocalDateTime.parse(column.getValue(), dateTimeFormatter));
//                default -> {
//                }
//            }
//        }
//
//        return user;
//    }
//
//    /**
//     * es批量新增、更新文档（不存在：新增， 存在：更新）
//     *
//     * @param userList 集合
//     * @throws IOException
//     */
//    private void index(List<Users> userList) throws IOException {
//        BulkRequest.Builder br = new BulkRequest.Builder();
//
//        userList.forEach(user -> br
//                .operations(op -> op
//                        .index(idx -> idx
//                                .index("users")
//                                .id(String.valueOf(user.getId()))
//                                .document(user))));
//
//        client.bulk(br.build());
//    }
//
//    /**
//     * es批量删除文档
//     *
//     * @param idList id集合
//     * @throws IOException
//     */
//    private void delete(List<String> idList) throws IOException {
//        BulkRequest.Builder br = new BulkRequest.Builder();
//
//        idList.forEach(id -> br
//                .operations(op -> op
//                        .delete(idx -> idx
//                                .index("users")
//                                .id(id))));
//
//        client.bulk(br.build());
//    }
//
//}
